package com.pai4j.common.service.messagequeue.producer;


import com.pai4j.common.util.RedisUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
/**
 *  Redis消息队列发送者
 *
 * @author: CYM-pai
 * @date: 2025/04/29 21:13
 **/
@Slf4j
@Service
public class RedisMessageQueueSender<T, P> extends AbstractMessageQueueSender<T, P> {

    @Resource
    private RedisUtil redisUtil;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 普通队列发送
     *
     * @param queue
     * @param message
     * @return
     */
    @Override
    protected boolean doSend(String queue, T message) {
        String messageValue = super.serializeMessage(message);
        redisUtil.lleftPush(queue, messageValue);
        return true;
    }
    
    /**
     * 广播发送
     *
     * @param channel
     * @param message
     * @return
     */
    @Override
    @Retryable(
            value = {RedisSystemException.class, Exception.class},
            maxAttempts = 5,
            backoff = @Backoff(delay = 2000, multiplier = 1.5, maxDelay = 10000)
    )
    protected boolean broadSend(String channel, T message) {
        try {
            String messageValue = super.serializeMessage(message);
            // Use StringRedisTemplate to ensure payload is published as plain UTF-8 string bytes
            stringRedisTemplate.convertAndSend(channel, messageValue);
            log.debug("Successfully broadcast message to channel: {}", channel);
            return true;
        } catch (Exception e) {
            log.warn("Failed to broadcast message to channel: {}, attempt will retry. Error: {}", channel, e.getMessage());
            throw e;
        }
    }
    /**
     * 优先级发送
     *
     * @param queue
     * @param message
     * @param priority :文章推送时间，推送时间距离当前时间越近（两者差值越小），则优先级越高
     * @return
     */
    @Override
    protected boolean prioritySend(String queue, T message, P priority) {
        // 先将消息序列化成json字符串
        String messageValue = super.serializeMessage(message);
        // 优先级发送:将优先级值作为分数，消息值作为值，保存到redis的zset集合中
        // zset集合的分数值越小，则优先级越高, queue->队列名称做为key
        redisUtil.zadd(queue, messageValue, (Long) priority, -1L);
        return true;
    }
}
